package com.swak.reactivex.transport;

import java.util.concurrent.CompletionStage;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.swak.exception.RemotingException;
import com.swak.reactivex.transport.exception.ExchangeException;
import com.swak.reactivex.transport.exception.ExchangeException.ExchangeMethod;

/**
 * 主要处理接收事件的处理
 * 
 * @author DELL
 */
public class HeaderExchangeHandler extends ChannelHandlerDelegate {

	protected static final Logger logger = LoggerFactory.getLogger(HeaderExchangeHandler.class);

	public HeaderExchangeHandler(ExchangeHandler handler) {
		super(handler);
	}

	@Override
	public void connected(Channel channel) throws RemotingException {
		ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
		handler.connected(exchangeChannel);
	}

	@Override
	public void disconnected(Channel channel) throws RemotingException {
		ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
		try {
			handler.disconnected(exchangeChannel);
		} finally {
			TimeoutFuture.closeChannel(channel);
			HeaderExchangeChannel.removeChannel(channel);
		}
	}

	@Override
	public void sent(Channel channel, Object message) throws RemotingException {
		Throwable exception = null;
		try {
			ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
			handler.sent(exchangeChannel, message);
		} catch (Throwable t) {
			exception = t;
			HeaderExchangeChannel.removeChannelIfDisconnected(channel);
		}
		if (message instanceof Message && ((Message) message).isRequest()) {
			Message request = (Message) message;
			TimeoutFuture.sent(channel, request);
		}

		// 这段异常该如何处理，并不能体现在执行的流程上
		if (exception != null) {
			logger.error("sent error {}", exception);
		}
	}

	@Override
	public void received(Channel channel, Object message) throws RemotingException {
		final ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
		if (Message.isResquest(message)) {
			Message request = (Message) message;
			if (request.isTwoWay()) {
				handleRequest(exchangeChannel, request);
			} else {
				handleReceived(exchangeChannel, request);
			}
		} else if (Message.isResponse(message)) {
			handleResponse(channel, (Message) message);
		} else {
			handleReceived(exchangeChannel, message);
		}
	}

	void handleReceived(final ExchangeChannel exchangeChannel, Object message) throws RemotingException {
		handler.received(exchangeChannel, message);
	}

	void handleResponse(Channel channel, Message response) throws RemotingException {
		if (response != null && !response.isHeartbeat()) {
			TimeoutFuture.received(channel, response);
		}
	}

	void handleRequest(final ExchangeChannel channel, Message req) throws RemotingException {
		try {
			CompletionStage<Object> future = ((ExchangeHandler) handler).reply(channel, req);
			future.whenComplete((appResult, t) -> {
				Message response = Message.forResponse(req);
				try {
					if (t == null) {
						response.setStatus(Message.OK);
						response.setData(
								(appResult != null && appResult instanceof Message) ? ((Message) appResult).getData()
										: appResult);
					} else {
						response.setStatus(Message.SERVICE_ERROR);
						response.setData(t.getMessage());
					}
					channel.send(response);
				} catch (RemotingException e) {
					logger.warn("Send result to consumer failed, channel is " + channel + ", msg is " + e);
				}
			});
		} catch (Throwable e) {
			Message response = Message.forResponse(req);
			response.setStatus(Message.SERVICE_ERROR);
			response.setData(e.getMessage());
			channel.send(response);
		}
	}

	/**
	 * write 过程中的异常
	 * 
	 * 1. 请求 -- 模拟一个响应 执行 received
	 * 2. 响应 -- 发送一个错误状态码的响应
	 * 
	 * read 过程中的异常
	 * 1. 请求 -- 发送一个错误状态码的响应
	 * 2. 响应 -- 模拟一个响应 执行received
	 * 
	 */
	@Override
	public void caught(Channel channel, Throwable exception) throws RemotingException {

		// 打印错误
		if (logger.isWarnEnabled()) {
			logger.warn("catch error：", exception);
		}

		// 执行异常
		ExchangeException e = null;
		if (exception instanceof ExchangeException) {
			e = (ExchangeException) exception;
		} else if (exception.getCause() != null && exception.getCause() instanceof ExchangeException) {
			e = (ExchangeException) exception.getCause();
		}

		// 处理请求和响应
		Message message = null;
		if (e != null && e.getData() != null && e.getData() instanceof Message
				&& (message = (Message) e.getData()) != null && !message.isHeartbeat()
				&& (message.isRequest() || message.isResponse())) {
			if (e.getMethod() == ExchangeMethod.Write) {
				if (message.isRequest()) {
					Message res = Message.forResponse(message);
					res.setStatus(Message.BAD_REQUEST);
					res.setError(e.getCause());
					this.received(channel, res);
				}

				// 防止陷于死循环
				else if (message.getStatus() != Message.BAD_RESPONSE) {
					Message res = Message.forResponse(message);
					res.setStatus(Message.BAD_RESPONSE);
					res.setError(e.getCause());
					channel.send(res);
				}

				// 发送一个空的响应过去
				else {
					Message res = Message.forResponse(message);
					res.setStatus(Message.BAD_RESPONSE);
					channel.send(res);
				}
			}

			if (e.getMethod() == ExchangeMethod.Read) {
				if (message.isRequest()) {
					Message res = Message.forResponse(message);
					res.setStatus(Message.BAD_REQUEST);
					res.setError(e.getCause());
					channel.send(res);
				} else {
					Message res = Message.forResponse(message);
					res.setStatus(Message.BAD_RESPONSE);
					res.setError(e.getCause());
					this.received(channel, res);
				}
			}
			return;
		}

		ExchangeChannel exchangeChannel = HeaderExchangeChannel.getOrAddChannel(channel);
		try {
			handler.caught(exchangeChannel, exception);
		} finally {
			HeaderExchangeChannel.removeChannelIfDisconnected(channel);
		}
	}
}
